package org.zjt.chapter1.monitor;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.buffer.Unpooled;
import io.netty.buffer.UnpooledDirectByteBuf;
import io.netty.util.internal.PlatformDependent;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.Field;
import java.nio.ByteBuffer;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Function;

/**
 * Description:
 *
 * @author juntao.zhang
 * Date: 2018-11-30 4:14 PM
 * @see io.netty.util.internal.PlatformDependent#DIRECT_MEMORY_COUNTER 来监控直接内存的大小
 */
public class NettyDirectMemoryMonitor {


    private static Function<ByteBuf,String> byteBuf2Str = byteBuf -> {
        byte[] bytes = new byte[byteBuf.readableBytes()];
        byteBuf.readBytes(bytes);
        return new String(bytes);
    };

    @Slf4j
    public static class DirectMemReporter {
        private AtomicLong directMem = new AtomicLong();
        private ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);

        public DirectMemReporter() throws Exception {
            Field field = PlatformDependent.class.getDeclaredField("DIRECT_MEMORY_COUNTER");
            field.setAccessible(true);
            try {
                directMem = (AtomicLong) field.get(PlatformDependent.class);
            } catch (IllegalAccessException e) {
                e.printStackTrace();
            }
        }

        public void startReport() {
            executor.scheduleAtFixedRate(() -> System.out.printf("netty direct memory size:%s M, max:%s M\n", directMem.get() / (1 << 20), PlatformDependent.maxDirectMemory() / (1 << 20)), 0, 1, TimeUnit.SECONDS);
        }


        public void shutdown(){
            executor.shutdown();
        }
    }


    public static void main(String[] args) throws Exception {
        DirectMemReporter directMemReporter = new DirectMemReporter();
        directMemReporter.startReport();
        TimeUnit.SECONDS.sleep(1);


        PooledByteBufAllocator allocator = new PooledByteBufAllocator(true);
        ByteBuf byteBuf = allocator.directBuffer(15);
        byteBuf.writeBytes("使用池化内存".getBytes());
        System.out.println("使用池化内存  " +byteBuf.getClass().toString()+"  中内容：" + byteBuf2Str.apply(byteBuf));


        TimeUnit.SECONDS.sleep(1);

        ByteBuf directBuffer = Unpooled.directBuffer(1024);
        directBuffer.writeBytes("jsdjfksdfsdkfjsajkdfhsjdhfkjs".getBytes());
        System.out.println("使用池化内存  " +directBuffer.getClass().toString()+"  中内容：" + byteBuf2Str.apply(directBuffer));


        TimeUnit.SECONDS.sleep(4);


        directBuffer = Unpooled.directBuffer(10 << 20);
        directBuffer.writeBytes("jsdjfksdfsdkfjsajkdfhsjdhfkjs".getBytes());
        System.out.println("使用池化内存  " +directBuffer.getClass().toString()+"  中内容：" + byteBuf2Str.apply(directBuffer));


        TimeUnit.SECONDS.sleep(4);
//        directMemReporter.shutdown();

    }

}
